package com.xzx.flink.wordcount;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2022-03-19 11:05 PM
 */
public class StreamWordCountV1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // DataStreamSource继承自DataStream
        DataStreamSource<String> textFile = env.readTextFile("flink1.14.4/src/main/resources/wordcount.txt");
        // SingleOutputStreamOperator继承自DataStream
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMap = textFile.flatMap((String value, Collector<Tuple2<String, Integer>> out) -> {
            String[] strings = value.split(" ");
            for (String string : strings) {
                out.collect(Tuple2.of(string, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));
        // KeyedStream<T, KEY>继承自DataStream，
        KeyedStream<Tuple2<String, Integer>, String> keyBy = flatMap.keyBy(value -> value.f0);
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = keyBy.sum(1);
        sum.print();

        // 实时jon，需要触发提交
        env.execute("WordCount");
    }
}
